package day06

import org.apache.spark.rdd.RDD
import org.apache.spark.{HashPartitioner, Partitioner, SparkConf, SparkContext}

/**
 * 2.9	键值对RDD数据分区
 *
 * Spark目前支持Hash分区和Range分区，和用户自定义分区。Hash分区为当前的默认分区。
 *
 * 分区器直接决定了RDD中分区的个数、RDD中每条数据经过Shuffle后进入哪个分区和Reduce的个数。
 *
 * 1）注意：
 * ①只有Key-Value类型的RDD才有分区器，非Key-Value类型的RDD分区的值是None
 * ②每个RDD的分区ID范围：0~numPartitions-1，决定这个值是属于那个分区的。
 *
 * 2）获取RDD分区
 */
object KeyValue_Rdd_Partition {

  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3 创建RDD
    val pairRDD: RDD[(Int, Int)] = sc.makeRDD(List((1, 1), (2, 2), (3, 3)))

    // 打印分区器
    val partitioner: Option[Partitioner] = pairRDD.partitioner
    println(partitioner)

    // 使用HashPartitioner对RDD进行重新分区
    val partitionRdd: RDD[(Int, Int)] = pairRDD.partitionBy(new HashPartitioner(partitions = 2))

    // 打印分区器
    val partitioner1: Option[Partitioner] = partitionRdd.partitioner
    println(partitioner1)

    //4.关闭连接
    sc.stop()
  }
}

/**
 * 2.9.1	Hash分区
 * HashPartitioner分区原理：对于给定的key，计算其hashCode，并除以分区的个数取余，如果余数小于0，则用余数+分区的个数（否则加0），最后
 * 返回的值就是这个Key所属的分区ID
 *
 * HashPartitioner分区的弊端：可能导致每个分区中的数据量不均匀，极端情况下会导致某些分区拥有RDD的全部数据
 */
object KeyValue_RDD_Hash {

}

/**
 * 2.9.2	Ranger分区
 * RangePartitioner作用：将一定范围内的数映射到某一个分区内，尽量保证每个分区中的数据量均匀，而且分区与分区之间是有序的，
 * 一个分区中的元素肯定都是比另一个分区内的元素小或者大，但是分区内的元素是不能保证顺序的。简单地说，就是将一定范围内的数
 * 映射到某一个分区内
 *
 * 实现过程为：
 * 第一步：先从整个RDD中采用水塘抽样算法，抽取出样本数据，将样本数据排序，计算出每个分区的最大的key值，形成一个Array[KEY]
 * 类型的数组变量rangeBounds；
 *
 * 第二步：判断key在rangeBounds中所处的范围，给出该key值在下一个RDD中的分区id下标，该分区器要求RDD中的KEY类型必须是可以
 * 排序的
 */
object KeyValue_RDD_Ranger {

}

/**
 * 2.9.3	自定义分区
 * 详见2.4.3.1。MyPartitioner
 */
object KeyValue_RDD_customPartitioner {

}